package yz.mr.OutPut;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 需求：
 *      读取ReduceJoin的结果数据，并对数据进行包装，对于大于450分的写入一个结果文件，小于450的写入另外一个结果文件
 *      对于使用之前的内容分析：
 *          对Mapper端需要对数据进行切分包装处理，对于大于450分的写入一个Reduce，小于450的写入另外一个Reduce
 *          可以使用分区，对于大于450分的给一个编号，小于450的写入另外一个编号，在Reduce中不用做任何处理，直接写出
 *
 *      这种方式的缺点：
 *          在该过程会产生大量的网络IO
 *
 *      优化：
 *          可以避免使用Reduce，可以自定义输出类操作：
 *              在自定义的输入类中，对于传入的数据进行判断，结果将大于450分的写入一个结果文件，小于450的写入另外一个结果文件
 */
public class OutputDriver {
    public static void main(String[] args) throws Exception{
        Configuration entries = new Configuration();
        Job job = Job.getInstance(entries);
        //3.配置job
        job.setJobName("OutputDriver");
        job.setJarByClass(OutputDriver.class);

        job.setMapperClass(OutputMapper.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(OutputStudent.class);  //这里的Mapper和Reduce类型要一致

        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(OutputStudent.class);

        /**
         * 自定义输出类
         * setOutputFormatClass要求传入一个OutputFormat的子类，该子类有许多，有三个抽象子类
         * 由于需求只要定义写出数据的逻辑，对于其他的没有要求，所以要实现getRecordWrite方法就行
         * 对于其他的子类抽象方法，可以从子类中继承，所以自定义的输出类可以继承FileOutputFormat
         */

        job.setOutputFormatClass(MyOutputFormat.class);

        //7.设置输入输出路径(这是本地的路径)  输入路径既可以指定路径也可以指定目录
        FileInputFormat.addInputPath(job,new Path("output/ReduceCount"));
        FileOutputFormat.setOutputPath(job,new Path("output/output"));
        //8.提交执行当前JOB
        job.waitForCompletion(true);
    }
}
